package com.cet.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class CreateDemo {

    public static void main(String[] args) throws IOException {

        // 同步
        Flux.generate(
                () -> 0,
                (state, sink) -> {
                    sink.next(state);
                    if (state == 10) {
                        sink.complete();
                    }
                    return state + 1;
                }
        ).log().subscribe();

        // 异步

        Flux<Integer> flux = Flux.create(sink -> {
            for (int i = 0; i < 1000; i++) {
                sink.next(i);
                // 模拟处理延迟
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                }
            }
            sink.complete();
        }, FluxSink.OverflowStrategy.DROP); // 使用DROP策略

        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(1); // 每次只请求1个元素
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println(value);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                }
                request(1); // 处理完再请求下一个
            }
        });

    }
}
